/*
 * Copyright 2011 Rolf Aden
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License. 
 */

package com.ec;

import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.tools.mapreduce.DatastoreMutationPool;
import com.google.appengine.tools.mapreduce.contrib.DatastoreMapper;

/**
 * Mapper for adding new proerties.
 * 
 * @author Rolf Aden
 */
public class AddPropertyMapper extends DatastoreMapper {
	
	enum Phase {
		run,
		verify
	}
	
	private static final Logger LOG = Logger.getLogger(AddPropertyMapper.class.getName());
	
	static final String PHASE = "ec.phase";
	static final String INITIAL_VALUE = "ec.initialValue";
	static final String PROPERTY_NAME = "ec.propertyName";
	static final String PROPERTY_TYPE = "ec.propertyType";
	static final String UNINDEXED = "ec.unindexed";
	
	private DatastoreService ds = DatastoreServiceFactory.getDatastoreService();
	private DatastoreMutationPool mutationPool; 
	
	private String propertyName;
	private Phase phase;
	private Object initialValue;
	private boolean unindexed;
	
	private ErrorReporter errorReporter;
	
	@Override
	public void taskSetup(Mapper<Key,Entity,NullWritable,NullWritable>.Context context) throws IOException, InterruptedException {
		super.taskSetup(context);

		this.propertyName = context.getConfiguration().get(PROPERTY_NAME);
		this.phase = "verify".equals(context.getConfiguration().get(PHASE))
			? Phase.verify
			: Phase.run;
		
		String propertyType = context.getConfiguration().get(PROPERTY_TYPE); 
		String value = context.getConfiguration().get(INITIAL_VALUE);
		this.initialValue = TypeConverter.parse(propertyType, value); 
		this.unindexed = Boolean.parseBoolean(context.getConfiguration().get(UNINDEXED));
		
		//	ask the appengine-mapreduce-java team how to communicate a fatal error, 
		//	meaning it would make no sense to proceed with the task, e.g. no property name given
		//	refer to comment in MapReduceServlet#handleMapperWorker, currently it's not possible
		validate();
		
		this.mutationPool = this.getAppEngineContext(context).getMutationPool();
		this.errorReporter = new ErrorReporter(context, ds);
	}

	@Override
	public void map(Key key, Entity entity, Mapper<Key,Entity,NullWritable,NullWritable>.Context context) throws IOException, InterruptedException 
	{
		if (LOG.isLoggable(Level.FINER)) {
			LOG.finer("processing " + entity + " in phase " + this.phase);
		}

		boolean isFinestLogLevel = LOG.isLoggable(Level.FINEST);

		switch (this.phase) {
			case run:
	    		if (!entity.hasProperty(this.propertyName))
	    		{
	    			if (this.unindexed) {
	    				entity.setUnindexedProperty(this.propertyName, this.initialValue);
	    			} else {
	    				entity.setProperty(this.propertyName, this.initialValue);
	    			}	
	    			this.mutationPool.put(entity);	
					if (isFinestLogLevel) {
						LOG.finest((this.unindexed ? "unindexed " : "") +
								this.propertyName + " with value "
								+ this.initialValue +
								" added to " + entity);
					}
	    		}
				break;

			case verify:
				if (!entity.hasProperty(this.propertyName)) {
					String errMsg = "Entity " + key + ": has no property " + this.propertyName;
					this.errorReporter.addErrorMessage(errMsg);
					if (isFinestLogLevel) {
						LOG.finest(errMsg);
					}
				} else {
					if ( !equals(this.initialValue, entity.getProperty(this.propertyName)) ) {
						String errMsg = "Entity " + key + ": " + this.propertyName + " has the wrong value!";
						this.errorReporter.addErrorMessage(errMsg);
						if (isFinestLogLevel) {
							LOG.finest(errMsg);
						}
					}
				}
				break;
		}
	}
	
	protected void validate() throws IOException {
		if (this.propertyName == null || this.propertyName.trim().isEmpty()) {
			throw new IOException("property name is empty");
		}
	}
	
	private boolean equals(Object o1, Object o2)
	{
		return o1 == null ? o2 == null : o1.equals(o2);
	}
}
